package xdu.lz.stage6_threadpool.chapter3_threadPoolExecutorAPI;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 *
 * JDK1.8 才有的，支持异步回调的链式编程的API
 *
 *
 * 解决的问题：
 * {@link ExecutorService}的
 * 1.批量处理任务的时候，不会按照先完成的先拿结果，解决
 * {@link CompletionService}
 * 2.get的时候会阻塞，
 * {@link CompletableFuture}解决了这个问题 ，在task进行了注册，完成后会发生回调
 *
 * 3.假设任务分两阶段，第一阶段从DB读数据，第二阶段处理DB数据。
 * 那么ExecutorService批处理的时候，必须所有第一阶段全处理完才会返回结果，才会执行第二阶段。
 *
 * 为什么不能第一阶段处理完一个就执行第二阶段呢？各自跑各自的。
 */
public class CompletableFuture1 {

    public static void main(String[] args) throws Exception {
        test2();
    }

    /**
     * 内部维护的都是守护线程，如果不用线程池管理，都是守护的
     * {@link CompletableFuture#runAsync(Runnable)}
     *
     * 利用线程池管理
     * {@link CompletableFuture#runAsync(Runnable, Executor)}
     *
     * 注册完成回调事件，是双参数的，第一个是返回值 ，第二个是异常
     * {@link CompletableFuture#whenComplete(BiConsumer)}
     */
    private static void test1(){
        ExecutorService service = Executors.newSingleThreadExecutor();
        CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },service).whenComplete(new BiConsumer<Void, Throwable>() {
            @Override
            public void accept(Void aVoid, Throwable throwable) {
                System.out.println("方法执行完毕，返回值是"+aVoid+" ,异常是"+throwable);
            }
        });

        System.out.println("=========not blocked");
        service.shutdown();
    }

    /**
     * 测试ExecutorService自带的批量处理任务缺点
     * invokeAll等所有第一阶段都完成
     * 1: done,return 1
     * 3: done,return 3
     * 4: done,return 4
     * 0: done,return 0
     * 2: done,return 2
     * 处理后的数据:1
     * 处理后的数据:2
     * 处理后的数据:3
     * 处理后的数据:4
     * 处理后的数据:5
     * 第二阶段任务结束
     */
    private static void test2() throws Exception {
        ExecutorService service = Executors.newFixedThreadPool(5);
        List<Callable<Integer>>  taskList = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int finalI = i;
            taskList.add(()->{
                int time = ThreadLocalRandom.current().nextInt(5);
                Thread.sleep(time*1000);
                System.out.println(finalI +": done,return "+finalI);
                return finalI;
            });
        }
        System.out.println("invokeAll等所有第一阶段都完成");
        //blocked util all the task done
        List<Future<Integer>> futureList = service.invokeAll(taskList);
        for (Future<Integer> future : futureList) {
            Integer integer = future.get();
            //对数据进行操作
            System.out.println("处理后的数据:"+(++integer));
        }
        System.out.println("第二阶段任务结束");

        service.shutdown();
    }

    /**
     *
     * {@link CompletableFuture#supplyAsync(Supplier, Executor)}
     *
     * 批量处理任务 ，哪个先完哪个就执行第二阶段
     * 0:done,value =  0
     * 0done and precess its data: = 1
     * 2:done,value =  2
     * 4:done,value =  4
     * 2done and precess its data: = 3
     * 4done and precess its data: = 5
     * 3:done,value =  3
     * 3done and precess its data: = 4
     * 1:done,value =  1
     * 1done and precess its data: = 2
     */
    private static void test3() throws Exception {
        ExecutorService service = Executors.newFixedThreadPool(5);
        List<Callable<Integer>>  taskList = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            int finalI = i;
            taskList.add(()->{
                int time = ThreadLocalRandom.current().nextInt(5);
                Thread.sleep(time*1000);
                System.out.println(finalI +": done,return "+finalI);
                return finalI;
            });
        }

        for (int i = 0; i < 5; i++) {
            final int finalI = i;
            CompletableFuture.supplyAsync(new Supplier<Integer>() {
                @Override
                public Integer get() {
                    int time = ThreadLocalRandom.current().nextInt(5);
                    try {
                        Thread.sleep(time * 1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return finalI;
                }
            }, service).thenAccept(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) {
                    System.out.println(integer + ":done,value =  " + integer);
                }
            }).whenComplete(new BiConsumer<Void, Throwable>() {
                @Override
                public void accept(Void aVoid, Throwable throwable) {
                    System.out.println(finalI + "done and precess its data: = " + (finalI + 1));
                }
            });
        }
        service.shutdown();
    }
}
